package com.atguigu.flink.wordcount;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/12
 *
 为接口提供实现：
 *          方式一： 提供外部类的实现
 *                  输入的数据:  水位传感器的id,ts,vc(水位值)
 *
 *          方式二： 提供内部类的实现
 *
 *
 */
public class Demo5_OuterClass
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop103",8888);

        ds.map(new WaterSensorMapFunction())
          .print();


    env.execute();


    }


    //内部类实现 只在当前类中有效
    private static class WaterSensorMapFunction implements MapFunction<String, WaterSensor>
    {
        @Override
        public WaterSensor map(String value) throws Exception {
            String[] words = value.split(",");
            return new WaterSensor(
                words[0],
                Long.valueOf(words[1]),
                Integer.valueOf(words[2])
            );
        }
    }
}
